/*
 * To change this license header, choose License Headers in Project Properties.
 * To change this template file, choose Tools | Templates
 * and open the template in the editor.
 */
package cn.ac.iie.di.ban.data.exchange.runner.slave.RocketMQHelper.SendMQ;

import cn.ac.iie.di.datadock.rdata.exchange.client.core.REFieldType;
import cn.ac.iie.di.datadock.rdata.exchange.client.exception.REConnectionException;
import cn.ac.iie.di.datadock.rdata.exchange.client.exception.RESessionException;
import cn.ac.iie.di.datadock.rdata.exchange.client.v1.REMessage;
import cn.ac.iie.di.datadock.rdata.exchange.client.v1.connection.REConnection;
import cn.ac.iie.di.datadock.rdata.exchange.client.v1.session.RESendSession;
import cn.ac.iie.di.datadock.rdata.exchange.client.v1.session.RESendSessionBuilder;
import cn.ac.iie.di.ban.data.exchange.runner.slave.RocketMQHelper.data.RocketMQArgvData;
import cn.ac.iie.di.ban.data.exchange.runner.slave.task.checkTask.CheckSQLAndPermissionTask;
import com.alibaba.fastjson.JSONArray;
import com.alibaba.fastjson.JSONObject;
import org.apache.avro.generic.GenericRecord;
import org.apache.commons.lang.exception.ExceptionUtils;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

import java.util.Arrays;
import java.util.HashMap;
import java.util.List;
import java.util.Map;
import java.util.logging.Level;

import static cn.ac.iie.di.datadock.rdata.exchange.client.core.REFieldType.Struct;
import static cn.ac.iie.di.datadock.rdata.exchange.client.core.REFieldType.Structs;

/**
 * @author Black-Laptop
 */
public class SendMQ {

    private static final Logger LOG = LoggerFactory.getLogger(SendMQ.class);
    private static REConnection conn;

    private RESendSessionBuilder builder;
    private RESendSession session = null;

    private List<String> colNames;
    private REFieldType[] types;

    private String sendConsumerNameserver;
    private String sendConsumerTopic;

    public SendMQ(REMessage message, RocketMQArgvData rocketMQArgvData) throws Exception {

        String[] columns = (String[]) changColumns(message.colNames);
        this.colNames = Arrays.asList(columns);

        this.types = message.types;
        this.sendConsumerNameserver = rocketMQArgvData.getSendConsumerNameserver();
        this.sendConsumerTopic = rocketMQArgvData.getSendConsumerTopic();
        conn = new REConnection(this.sendConsumerNameserver);
        initBuilder();
        addColumns(message.data.get(0));
    }


    /**
     * 将发送队列中的列名转换回去
     *
     * @param columns
     * @return
     */
    private Object[] changColumns(Object[] columns) {
        JSONObject jsonPostResult = JSONObject.parseObject(CheckSQLAndPermissionTask.postResult);
        JSONObject jsonDataResult = (JSONObject) jsonPostResult.get("data");
        JSONArray array = jsonDataResult.getJSONArray("cols");
//        System.out.println(array.toString());

        for (int j = 0; j < columns.length; j++) {
            for (int i = 0; i < array.size(); i++) {
                JSONObject jsonColsResult = (JSONObject) array.get(i);
//                System.out.println("cols:"+jsonColsResult.get("srccolumn"));
                if (columns[j].equals(jsonColsResult.get("srccolumn"))) {
                    columns[j] = jsonColsResult.get("showcolumn");
                }
            }
        }
        return columns;
    }

    public void sendMessage(List<String> colNames, List<Object[]> datas) throws Exception {
        //生成builder，参数为topic，由具体业务决定

        /**
         * 这里是设置struct或者structs的格式。 首先需要addColumn
         * 其次需要在对应的类型里面setDetail，参数是：列名，子列类型map
         */
        //开始发送。
        //这里，DocSchemaName和UserDesc非必须，具体什么时候需要，由具体业务决定。
        //DocSchemaName设置一次后不会清空，每次flush都是同一个，如果需要改变，需要再次设置。
        //for (int r = 0; r < 10; ++r) {
        //添加数据。set完所有列后add即可添加一条数据。多条数据打一包发送。
        //各种set方法可以通过session.set补全来查看。set方法和REFieldType各类对应。
        //注意：尽量不要一条数据发送一次，这样会带来较大的额外开销
        //注意2：本地缓存数据量有大小限制，如果超过500M，会报错。
        //注意3：如果要用set（index，value）的方式设置数据，index从1开始。
        LOG.debug("datas size: " + datas.size());
        LOG.debug("type size: " + types.length);

        for (Object[] data : datas) {
            addSession(data);
        }
        //添加完所有数据，flush即可发送。
        session.flush();
        LOG.info("Send Message " + (datas.size()));
    }

    public void stop() throws Exception {
        try {
            session.shutdown();
        } catch (REConnectionException e) {
            LOG.error(ExceptionUtils.getFullStackTrace(e));
            throw new Exception(ExceptionUtils.getFullStackTrace(e));
        }
    }


    private void initBuilder() throws Exception {
        try {
            builder = (RESendSessionBuilder) conn.getSendSessionBuilder(this.sendConsumerTopic);
        } catch (REConnectionException e) {
            LOG.error(ExceptionUtils.getFullStackTrace(e));
            throw new Exception(ExceptionUtils.getFullStackTrace(e));
        }
    }

    private void addColumns(Object[] datas) {
        //设置数据结构，参数为每一列的列名、列类型、是否可以为空。注意： 如果设置为不能为空，那么如果添加数据时没有添加该列数据，会报错。
        for (int i = 0; i < colNames.size(); i++) {
            builder.addColumn(colNames.get(i), types[i], true);
            setDetail(colNames.get(i), types[i], datas[i]);

        }
        LOG.debug("colName: " + Arrays.toString(colNames.toArray()) + "  type: " + Arrays.toString(types));
        //生成连接，参数为rocketmq的nameserver串。可以多个nameserver，分号隔开。由具体业务决定
        LOG.info("topic: " + this.sendConsumerTopic);
    }

    private void setDetail(String column, REFieldType type, Object data) {
        switch (type) {
            case Struct:
                setStructDetail(column, data);
                break;
            case Structs:
                setStructsDetail(column, data);
                break;
        }
    }

    private void setStructDetail(String column, Object data) {
        HashMap<String, REFieldType> detailMap = new HashMap<>();
        for (Object key : ((Map) data).keySet()) {
            detailMap.put((String) key, GetVarType(((Map) data).get(key)));
            LOG.debug("Key = " + key);
        }
        Struct.addDetail(column, detailMap);
    }

    private void setStructsDetail(String column, Object data) {
        HashMap<String, REFieldType> detailMap = new HashMap<>();
        Map[] mapArray = data instanceof Map[] ? (Map[]) data : null;
        if (mapArray != null) {
            for (int indexMap = 0; indexMap < mapArray.length; indexMap++) {
                for (Object key : mapArray[indexMap].keySet()) {
                    detailMap.put((String) key, GetVarType(mapArray[indexMap].get(key)));
                    LOG.debug("Key = " + key);
                }
            }

            Structs.addDetail(column, detailMap);
        } else {
            Structs.addDetail(column, detailMap);
            if (data == null) {
                LOG.error("data value: " + data + " is not structs");
            } else {
                LOG.error("data value: " + data + " is not structs, is " + data.getClass());
            }
        }
    }

    private static REFieldType GetVarType(Object object) {
        String type = object.getClass().getSimpleName();
        REFieldType fieldType = null;
        switch (type) {
            case "String": {
                fieldType = REFieldType.String;
                break;
            }
            case "Integer": {
                fieldType = REFieldType.Int;
                break;
            }
            case "Long": {
                fieldType = REFieldType.Long;
                break;
            }
            default: {
                LOG.error("type not found: " + type);
            }
        }
        return fieldType;
    }

    public void start() throws Exception {
        int retryTime = 0;
        while (true) {
            try {
                session = (RESendSession) builder.build();
                session.start();
                LOG.info("session start successfully. retry time: " + retryTime);
                break;
            } catch (RESessionException | REConnectionException e) {
                retryTime++;
                LOG.error(ExceptionUtils.getFullStackTrace(e));
                stop();
                try {
                    Thread.sleep(1000);
                } catch (InterruptedException ex) {
                    java.util.logging.Logger.getLogger(SendMQ.class.getName()).log(Level.SEVERE, null, ex);
                }
            }
        }
    }

    public void sendMessage(REMessage message) throws Exception {
        List<String> colNames = Arrays.asList(message.colNames);
//        System.out.println("============================colNames:"+colNames.toString());

        List<Object[]> datas = message.data;
        try {
            sendMessage(colNames, datas);
        } catch (RESessionException | REConnectionException e) {
            LOG.error(ExceptionUtils.getFullStackTrace(e));
            throw new Exception(ExceptionUtils.getFullStackTrace(e));
        }
    }


    private void addSession(Object[] data) throws Exception {

        LOG.debug("send data: " + Arrays.toString(data));
        for (int i = 0; i < types.length; i++) {
            setSession(colNames.get(i), types[i], data[i]);

        }

        try {
            session.add();
        } catch (Exception e) {
            throw new Exception(ExceptionUtils.getFullStackTrace(e));
        }
    }

    private void setSession(String column, REFieldType type, Object data) throws Exception {
        try {
            if (data == null || data.equals("null") || data.equals("\"null\"")) {
                return;
            }

            LOG.debug("column: " + column + " type:" + type + " value: " + data);

            switch (type) {
                case String:
                    session.setString(column, String.valueOf(data));
                    break;
                case Int:
                    session.setInt(column, Integer.valueOf(data.toString()));
                    break;
                case Long:
                    session.setLong(column, Long.valueOf(data.toString()));
                    break;
                case Float:
                    session.setFloat(column, Float.valueOf(data.toString()));
                    break;
                case Double:
                    session.setDouble(column, Double.valueOf(data.toString()));
                    break;
                case Binary:
                    session.setBinary(column, data.toString().getBytes());
                    break;
                case Doc:
                    session.setDoc(column, (GenericRecord) data);
                    break;
                case Docs:
                    session.setDocs(column, (GenericRecord[]) data);
                    break;
                case Binaries:
                    session.setBinaries(column, (byte[][]) data);
                    break;
                case Doubles:
                    session.setDoubles(column, (double[]) data);
                    break;
                case Floats:
                    session.setFloats(column, (float[]) data);
                    break;
                case Longs:
                    session.setLongs(column, (long[]) data);
                    break;
                case Ints:
                    session.setInts(column, (int[]) data);
                    break;
                case Strings:
                    session.setStrings(column, (String[]) data);
                    break;
                case Boolean:
                    session.setBoolean(column, (boolean) data);
                    break;
                case Booleans:
                    session.setBooleans(column, (boolean[]) data);
                    break;
                case Struct:
                    session.setStruct(column, (Map) data);
                    break;
                case Structs:
                    session.setStructs(column, (Map[]) data);
                    break;
                default:
                    throw new Exception("There is not the type of " + type);
            }
        } catch (Exception e) {
            throw new Exception(ExceptionUtils.getFullStackTrace(e));
        }
    }


}
